package net.xmmpp.uc.im.internal.web.config;


import net.xmmpp.uc.im.common.model.MsgLog;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.queryablestate.client.QueryableStateClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.net.UnknownHostException;

@Component
@EnableConfigurationProperties(FlinkProperties.class)
public class ApplicationConfig {

    private static final Logger LOGGER = LoggerFactory.getLogger(ApplicationConfig.class);


    @Autowired
    private FlinkProperties flinkProperties;

    @Bean(name = "configuration")
    public Configuration configuration(){
        Configuration config = new Configuration();
        config.setString(RestOptions.ADDRESS, flinkProperties.getJobHost());
        config.setInteger(RestOptions.PORT, flinkProperties.getJobPort());

        return config;
    }

    @Bean
    public QueryableStateClient queryableStateClient(){
        QueryableStateClient client = null;

        try {
            client = new QueryableStateClient(flinkProperties.getStateHost(), flinkProperties.getStatePort());
        } catch (UnknownHostException e) {
            LOGGER.error("QueryableStateClient Init Error", e);
        }

        return client;
    }

    @Bean(name = "msgLogListDescriptor")
    public ListStateDescriptor<MsgLog> msgLogListDescriptor(){
        ListStateDescriptor<MsgLog> descriptor = new ListStateDescriptor<>(
                "msgLogListSD",
                TypeInformation.of(new TypeHint<MsgLog>() {}).createSerializer(new ExecutionConfig()));

        return descriptor;
    }

    @Bean(name = "logSizeDescriptor")
    public ValueStateDescriptor<Integer> logSizeDescriptor(){
        ValueStateDescriptor<Integer> logSizeDescriptor = new ValueStateDescriptor<>("logSizeSD", Integer.class);

        return logSizeDescriptor;
    }


}
